package com.example.mq.mqserver.core;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.*;

/**
 * 通过这个类，实现消费的核心逻辑
 */
@Slf4j
public class ConsumerManager {
    //  持有 VirtualHost 引用，后续调用方法
    private final VirtualHost parent;
    //  引入线程池执行消费过程，设置固定线程数为4
    private final ExecutorService workPool = Executors.newFixedThreadPool(4);
    //  引入阻塞队列，存放队列名
    //  当有消息时，把消息所在的队列加入到阻塞队列中
    private final BlockingQueue<String> tokens = new LinkedBlockingQueue<>();
    //  创建一个扫描线程
    private final Thread scannerThread;
    //  存放所有消费者
    private ConcurrentHashMap<String,Consumer> consumerMap = new ConcurrentHashMap<>();

    public ConsumerManager(VirtualHost parent) {
        this.parent = parent;
        //  扫描线程死循环扫描令牌队列是否有新消息产生
        scannerThread = new Thread(() -> {
            while(true) {
                try {
                    String queueName = tokens.take();
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("扫描到的令牌不可用! ");
                    }

                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        //  设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    public void addConsumer(String consumerTag,String queueName,boolean autoAck,Consume consume) throws MqException {
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(queue == null) {
            throw new MqException("队列不存在! ");
        }

        Consumer consumer = new Consumer(consumerTag,queueName,autoAck,consume);
        synchronized (queue) {
            queue.addConsumer(consumer);
            consumerMap.put(consumerTag,consumer);
            //  此时可能已经有消息了，要把前面的消息给消费了
            int size = parent.getMemoryDataCenter().getQueueSize(queueName);
            for(int i = 0;i < size;i++) {
                consumeMessage(queue);
            }
        }

    }

    public void notifyConsumer(String queueName) throws InterruptedException {
        tokens.put(queueName);
    }

    public Consumer getConsumer(String consumerTag) {
        return consumerMap.get(consumerTag);
    }
    public void removeConsumer(String consumerTag) {
        consumerMap.remove(consumerTag);
    }


    private void consumeMessage(MSGQueue queue) {
        //  选出消费者
        Consumer consumer = queue.chooseConsumer();
        if(consumer == null) {
            return;
        }
        //  获取消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null) {
            return;
        }
        //  提交线程，执行消费
        workPool.submit(() -> {
            try {
                //  先把待确认消息保存起来
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //  执行回调
                consumer.getConsume().handleDelivery(consumer.getConsumerTag(),message.getBasicProperties(),message.getBody());
                //  自动应答,删除数据
                if(consumer.isAutoAck()) {
                    if(message.getDeliverMode() == 0x2) {
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    log.info("消息已自动应答.");
                }

                log.info("消费消息成功. queueName=" + queue.getName() + ",messageId=" + message.getMessageId());
            } catch (IOException | ClassNotFoundException | MqException e) {
                log.warn("消费消息失败! queueName=" + queue.getName() + ",messageId=" + message.getMessageId());
                e.printStackTrace();
            }

        });

    }

}
